package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
	"time"
)

var client sarama.SyncProducer

type logData struct{
	topic string
	data string
}
var logChan chan *logData

func Init(address []string,maxSize int) (err error){
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true

	client,err=sarama.NewSyncProducer(address,config)

	//初始化一个通道
	logChan=make(chan *logData,maxSize)
	go SendMsg()
	return
}

func SendToChan(topic,data string){
	msg :=&logData{
		topic:topic,
		data:data,
	}
	logChan<-msg
}

func SendMsg() {
	for{
		select{
			case ld:=<-logChan:
				msg := &sarama.ProducerMessage{}
				msg.Topic = ld.topic
				msg.Value = sarama.StringEncoder(ld.data)

				pid,offset,err:=client.SendMessage(msg)
				if err!=nil{
					fmt.Println("send message error",err)
					return
				}
				fmt.Printf("pid:%v offset:%v \n",pid,offset)
			default:
				time.Sleep(time.Millisecond*50)
		}
	}

}
